package db.split.app

import com.cw.realtime.common.Syntax.Pipe
import com.cw.realtime.common.base.FlinkRunner.{RunnerConfig, run}
import com.cw.realtime.common.constant.Constant._
import com.cw.realtime.common.util.CirceUtil.JsonObjOps
import com.cw.realtime.common.util.{DateFormatUtil, FlinkSinkUtil}
import io.circe.syntax.EncoderOps
import io.circe.{Json, JsonObject, parser}
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.{KeyedProcessFunction, ProcessFunction}
import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, OutputTag}
import org.apache.flink.util.Collector

import java.time.Duration


object DwdBaseLog {


  def main(args: Array[String]): Unit = {

    implicit val conf = RunnerConfig("dwd_base_log", TOPIC_LOG, 10011)

    run { (env, ds) =>

      //      ds.print()
      // 过滤掉错误的json数据
      val jsonObjStream = etl(ds)

      // 新老访客修复
      val keyedStream = keyByWithWatermark(jsonObjStream)
      val isNewFixed = fixNewVisitor(keyedStream)

//      isNewFixed.print()

      // 拆分日志, 侧输出流: page为主流
      // 启动页： start， err
      // 内容页： page，action，display(曝光), err
      val errTag = OutputTag[String]("error")
      val startTag = OutputTag[String]("start")
      val actionTag = OutputTag[String]("action")
      val displayTag = OutputTag[String]("display")
      val pageStream = splitLogStream(isNewFixed, errTag, startTag, actionTag, displayTag)
      val errStream = pageStream.getSideOutput(errTag)
      val startStream = pageStream.getSideOutput(startTag)
      val displayStream = pageStream.getSideOutput(displayTag)
      val actionStream = pageStream.getSideOutput(actionTag)

      //      pageStream.print("page")
      //      errStream.printToErr("err")
      //      actionStream.printToErr("action")
      //      displayStream.printToErr("display")
      //      startStream.printToErr("start")
      pageStream.sinkTo(FlinkSinkUtil.getKafkaSink(TOPIC_DWD_TRAFFIC_PAGE))
      errStream.sinkTo(FlinkSinkUtil.getKafkaSink(TOPIC_DWD_TRAFFIC_ERR))
      actionStream.sinkTo(FlinkSinkUtil.getKafkaSink(TOPIC_DWD_TRAFFIC_ACTION))
      displayStream.sinkTo(FlinkSinkUtil.getKafkaSink(TOPIC_DWD_TRAFFIC_DISPLAY))
      startStream.sinkTo(FlinkSinkUtil.getKafkaSink(TOPIC_DWD_TRAFFIC_START))

    }


  }

  /**
   * 数据清洗
   *
   * @param ds
   * @return
   */
  def etl(ds: DataStream[String]): DataStream[JsonObject] = {

    ds.flatMap { str =>
      for {
        json <- parser.parse(str).toOption // json格式错误的数据自动过滤
        jsonObj <- json.asObject
        common <- jsonObj("common")
        commonObj <- common.asObject
        mid <- commonObj("mid") if mid != Json.Null
        ts <- jsonObj("ts") if ts != Json.Null
      } yield {
        (jsonObj("page"), jsonObj("start")) match {
          case (None, None) | (Some(Json.Null), Some(Json.Null)) => null
          case _ => jsonObj
        }
      }
    }
  }

  /**
   * 设定事件时间和水印，返回kv stream
   */
  def keyByWithWatermark(jsonObjStream: DataStream[JsonObject]): KeyedStream[JsonObject, String] = {
    jsonObjStream.assignTimestampsAndWatermarks(
      WatermarkStrategy.forBoundedOutOfOrderness[JsonObject](
        Duration.ofSeconds(3L)).withTimestampAssigner(new SerializableTimestampAssigner[JsonObject] {
        override def extractTimestamp(element: JsonObject, recordTimestamp: Long): Long = {
          (for {
            ts <- element("ts").get.asNumber
            ts <- ts.toLong
          } yield ts).get
        }
      })
    ).keyBy {
      _.apply("common").get.asObject.get("mid").get.asString.get
    }
  }


  def fixNewVisitor(ds: KeyedStream[JsonObject, String]) = {
    ds.process(new KeyedProcessFunction[String, JsonObject, JsonObject] {
      type Context = KeyedProcessFunction[String, JsonObject, JsonObject]#Context
      var firstLoginState = Option.empty[ValueState[String]]

      override def open(parameters: Configuration): Unit = {
        firstLoginState = new ValueStateDescriptor("first_login_dt", classOf[String]) |>
          getRuntimeContext.getState[String] |>
          Option.apply[ValueState[String]]
      }

      override def processElement(value: JsonObject, ctx: Context, out: Collector[JsonObject]): Unit = {
        for {
          state <- firstLoginState
          common <- value("common").get.asObject
          ts <- value("ts").get.asNumber.get.toLong
          isNew = common("is_new").get.asString.get
          firstLoginDt = state.value//获取状态中记录的首次登录日期
          loginDt = DateFormatUtil.tsToDate(ts)
        } {
          if ("1" == isNew) {
            if (firstLoginDt == null) {
              // 新用户，和伪装成新用户且之前没被flink记录过的老用户
              state.update(loginDt)
              out.collect(value)
            }
            else if (loginDt != firstLoginDt) { // 不是真正的新访客
              val newValue = value.update("is_new", "0".asJson)
              out.collect(newValue)
            } else {
              out.collect(value)
              // 新用户重复登录
            }
          } else if ("0" == isNew && firstLoginDt == null) {
            // 还没有被flink记录过的老用户
            state.update(DateFormatUtil.tsToDate(ts - 24 * 60 * 60 * 1000))
          } else {
            // 老用户重复登录，或者is_new字段错误
          }
        }
      }
    })
  }


  def splitLogStream(ds: DataStream[JsonObject], errTag: OutputTag[String], startTag: OutputTag[String], actionTag: OutputTag[String], displayTag: OutputTag[String]): DataStream[String] = {
    ds.process(new ProcessFunction[JsonObject, String] {
      type Context = ProcessFunction[JsonObject, String]#Context
      val baseAttrs = Array("common", "ts") // 因为前面的过滤，这两个属性必有

      override def processElement(value: JsonObject, ctx: Context, out: Collector[String]): Unit = {
        //
        val baseJson = value.filter { tp => baseAttrs.contains(tp._1) }
        // 单独抽取err属性
        for (err <- value("err")) {
          ctx.output(errTag, err.toString)
        }
        // 抽取start
        for (start <- value("start")) {
          ctx.output(startTag, baseJson.add("start", start).toJson.toString)
        }
        value("page") match {
          case None =>
          case Some(page) =>
            val tmpJson = baseJson.add("page", page)
            // 抽取action数组
            for {
              actions <- value("actions")
              actions <- actions.asArray
              action <- actions
              action <- action.asObject
            } {
              ctx.output(actionTag, (action ++ tmpJson).toString)
            }
            // 抽取display曝光数组
            for {
              displays <- value("displays")
              displays <- displays.asArray
              display <- displays
              display <- display.asObject
            } {
              ctx.output(displayTag, (display ++ tmpJson).toString)
            }
            // page流
            out.collect(tmpJson.toJson.toString)
        }
      }
    })
  }


}




